package io.kiki.sba.registry.server.clientInterface.push;


import io.kiki.sba.registry.common.model.DataCenterPushInfo;
import io.kiki.sba.registry.common.model.SegmentPushInfo;
import io.kiki.sba.registry.common.model.SubscriberUtils;
import io.kiki.sba.registry.common.model.store.BaseInfo;
import io.kiki.sba.registry.common.model.store.DataCenterToSubDatumMap;
import io.kiki.sba.registry.common.model.store.PushData;
import io.kiki.sba.registry.common.model.store.Subscriber;
import io.kiki.sba.registry.core.model.Scope;
import io.kiki.sba.registry.trace.TraceId;
import lombok.Getter;
import lombok.Setter;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.Map.Entry;

public abstract class PushTask {
    protected final TraceId traceId;
    protected final DataCenterToSubDatumMap dataCenterToSubDatumMap;
    protected final Map<String, Subscriber> registerIdToSubscriberMap;
    protected final Subscriber subscriber;
    protected final PushingTaskKey pushingTaskKey;
    protected final Trace trace;
    protected volatile long expireTimestamp;
    protected int retryCount;

    @Setter
    @Getter
    private Map<String, DataCenterPushInfo> dataCenterPushInfos;

    protected PushTask(Cause cause, InetSocketAddress inetSocketAddress, Map<String, Subscriber> registerIdToSubscriberMap, DataCenterToSubDatumMap dataCenterToSubDatumMap) {
        this.traceId = TraceId.newTraceID();
        this.dataCenterToSubDatumMap = dataCenterToSubDatumMap;
        this.registerIdToSubscriberMap = registerIdToSubscriberMap;
        this.subscriber = registerIdToSubscriberMap.values().iterator().next();
        this.trace = Trace.trace(dataCenterToSubDatumMap, inetSocketAddress, subscriber.getAppName(), cause, registerIdToSubscriberMap.size(), SubscriberUtils.getMinRegisterTimestamp(registerIdToSubscriberMap.values()));
        this.pushingTaskKey = new PushingTaskKey(subscriber.getDataInfoId(), inetSocketAddress, subscriber.getScope(), subscriber.getClientVersion());
    }

    protected abstract boolean commit();

    protected abstract PushData createPushData();

    protected void expireAfter(long intervalMs) {
        this.expireTimestamp = System.currentTimeMillis() + intervalMs;
    }

    public boolean hasPushed() {
        if (registerIdToSubscriberMap.size() == 1) {
            return subscriber.hasPushed();
        }
        for (Subscriber subscriber : registerIdToSubscriberMap.values()) {
            if (!subscriber.hasPushed()) {
                return false;
            }
        }
        return true;
    }

    public boolean isSingletonReg() {
        return trace.cause.type == Type.reg && registerIdToSubscriberMap.size() == 1;
    }

    protected boolean afterThan(PushTask pushTask) {
        if (isSingletonReg() && pushTask.isSingletonReg()) {
            return subscriber.getVersion() > pushTask.subscriber.getVersion();
        }
        Set<Entry<String, Long>> entrySet = dataCenterToSubDatumMap.getDataCenterToSubDatumVersionMap().entrySet();
        for (Entry<String, Long> entry : entrySet) {
            // return true if one of any datum.version > t.datum.version
            if (entry.getValue() > pushTask.dataCenterToSubDatumMap.getDataCenterToSubDatumVersionMap(entry.getKey())) {
                return true;
            }
        }
        return false;
    }


    public Map<String, Integer> getDataCenterPushCount() {
        if (dataCenterPushInfos == null) {
            return Collections.EMPTY_MAP;
        }
        Map<String, Integer> pushDataCount = new HashMap<>(dataCenterPushInfos.size());
        for (Entry<String, DataCenterPushInfo> entry : dataCenterPushInfos.entrySet()) {

            int dataCenterPushCount = 0;
            if (entry.getValue().getSegmentPushInfos() != null) {
                dataCenterPushCount = entry.getValue().getSegmentPushInfos().values().stream().mapToInt(SegmentPushInfo::getDataCount).sum();
            }
            pushDataCount.put(entry.getKey(), dataCenterPushCount);
        }
        return pushDataCount;
    }

    public Map<String, Map<String, Integer>> getPushDataCount() {
        if (dataCenterPushInfos == null) {
            return Collections.EMPTY_MAP;
        }

        Map<String, Map<String, Integer>> pushDataCount = new HashMap<>(dataCenterPushInfos.size());
        for (Entry<String, DataCenterPushInfo> entry : dataCenterPushInfos.entrySet()) {
            Map<String, Integer> map = pushDataCount.computeIfAbsent(entry.getKey(), k -> new HashMap<>());
            if (entry.getValue().getSegmentPushInfos() == null) {
                continue;
            }
            for (SegmentPushInfo info : entry.getValue().getSegmentPushInfos().values()) {
                map.put(info.getSegment(), info.getDataCount());
            }
        }
        return pushDataCount;
    }

    protected static final class PushingTaskKey {
        protected final InetSocketAddress inetSocketAddress;
        protected final String dataInfoId;
        protected final Scope scope;
        protected final BaseInfo.ClientVersion clientVersion;

        protected PushingTaskKey(String dataInfoId, InetSocketAddress inetSocketAddress, Scope scope, BaseInfo.ClientVersion clientVersion) {
            this.dataInfoId = dataInfoId;
            this.inetSocketAddress = inetSocketAddress;
            this.scope = scope;
            this.clientVersion = clientVersion;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o)
                return true;
            if (o == null || getClass() != o.getClass())
                return false;
            PushingTaskKey that = (PushingTaskKey) o;
            return Objects.equals(inetSocketAddress, that.inetSocketAddress) && Objects.equals(dataInfoId, that.dataInfoId) && scope == that.scope && clientVersion == that.clientVersion;
        }

        @Override
        public int hashCode() {
            return Objects.hash(inetSocketAddress, dataInfoId, scope, clientVersion);
        }


    }
}
